package konggao.ots;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.zip.Deflater;

import com.aliyun.common.comm.ExecutionContext;
import com.aliyun.common.comm.RequestMessage;
import com.aliyun.common.comm.ResponseMessage;
import com.aliyun.common.utils.BinaryUtil;
import com.aliyun.common.utils.CodingUtils;
import com.aliyun.openservices.ClientException;
import com.aliyun.openservices.HttpMethod;
import com.aliyun.openservices.ots.internal.OTSCompressUtil;
import com.aliyun.openservices.ots.internal.OTSContentMD5ResponseHandler;
import com.aliyun.openservices.ots.internal.OTSRequestSigner;
import com.aliyun.openservices.ots.internal.OTSRetryStrategyFactroy;
import com.aliyun.openservices.ots.internal.OTSUtil;
import com.aliyun.openservices.ots.model.OTSProtocolHelper;
import com.aliyun.openservices.ots.model.PrimaryKeyRange;
import com.aliyun.openservices.ots.model.RangeRowQueryCriteria;

/**
 * @author whua
 * @email smtp2006@126.com
 * @version 2014年3月29日上午12:18:01
 */
public class OTSProxy
{

    private OTSConfig config;

    public String getRowsByRangeAsString( RangeRowQueryCriteria criteria, String transactionId, String nextToken )
        throws Exception
    {
        CodingUtils.assertParameterNotNull( criteria, "criteria" );

        if ( criteria.getRange() == null )
        {
            throw new IllegalArgumentException( OTSUtil.OTS_RESOURCE_MANAGER.getString( "MustSetCriteriaRange" ) );
        }

        boolean isReverse = criteria.getIsReverse();
        if ( ( !isReverse ) && ( PrimaryKeyRange.INF_MAX.equals( criteria.getRange().getBegin() ) ) )
        {
            throw new IllegalArgumentException( OTSUtil.OTS_RESOURCE_MANAGER.getString( "BeginIsInfMax" ) );
        }
        if ( ( isReverse ) && ( PrimaryKeyRange.INF_MIN.equals( criteria.getRange().getBegin() ) ) )
        {
            throw new IllegalArgumentException( OTSUtil.OTS_RESOURCE_MANAGER.getString( "BeginIsInfMin" ) );
        }

        int comp = OTSUtil.compare( criteria.getRange().getBegin(), criteria.getRange().getEnd() );
        if ( ( ( !isReverse ) && ( comp > 0 ) ) || ( ( isReverse ) && ( comp < 0 ) ) )
        {
            throw new IllegalArgumentException( OTSUtil.OTS_RESOURCE_MANAGER.getString( "BeginEndIncorrect" ) );
        }
        byte[] content = OTSProtocolHelper.buildGetRowsByRangeRequest( criteria, transactionId, nextToken );
        RequestMessage request = buildRequest( config.getEndpoint(), "GetRowsByRange", HttpMethod.POST, null, content );
        ResponseMessage obj = config.getClient().sendRequest( request, createContext( "GetRowsByRange", null ) );
        return read( obj.getContent() );
    }

    private String read( InputStream input )
        throws Exception
    {
        byte[] buffer = new byte[1024];
        ByteArrayOutputStream output = new ByteArrayOutputStream();
        int len = 0;
        while ( ( len = input.read( buffer ) ) > 0 )
        {
            output.write( buffer, 0, len );
        }
        return output.toString();
    }

    @SuppressWarnings( { "unchecked", "rawtypes" } )
    private RequestMessage buildRequest( URI endpoint, String action, HttpMethod httpMethod,
                                         Map<String, String> parameters, byte[] content )
    {

        if ( parameters == null )
        {
            parameters = new LinkedHashMap();
        }
        if ( content == null )
        {
            content = new byte[0];
        }

        RequestMessage request = new RequestMessage();
        request.setMethod( httpMethod );
        request.setEndpoint( endpoint );
        request.setResourcePath( action );
        request.setParameters( parameters );

        byte[] dataToSend = null;
        if ( config.getServiceConfig().isEnableRequestCompression() )
        {
            try
            {
                dataToSend = OTSCompressUtil.compress( new ByteArrayInputStream( content ), new Deflater() );
            }
            catch ( IOException e )
            {
                throw new ClientException(
                                           OTSUtil.OTS_RESOURCE_MANAGER.getFormattedString( "RequestCompressFail",
                                                                                            new Object[] { e.getMessage() } ) );
            }
            request.addHeader( "x-ots-request-compress-type", "deflate" );
            request.addHeader( "x-ots-request-compress-size", Integer.toString( content.length ) );
        }
        else
        {
            dataToSend = content;
        }

        request.setContent( new ByteArrayInputStream( dataToSend ) );
        request.setContentLength( dataToSend.length );

        addRequiredHeaders( request, dataToSend );

        return request;
    }

    private void addRequiredHeaders( RequestMessage request, byte[] content )
    {
        request.addHeader( "x-ots-contentmd5", BinaryUtil.toBase64String( BinaryUtil.calculateMd5( content ) ) );
        request.addHeader( "x-ots-apiversion", "2013-05-10" );
        if ( config.getServiceConfig().isEnableResponseCompression() )
            request.addHeader( "x-ots-response-compress-type", "deflate" );
    }

    private ExecutionContext createContext( String otsAction, Map<String, String> parameters )
    {
        ExecutionContext ec = new ExecutionContext();
        ec.setCharset( "utf-8" );
        ec.setSigner( new OTSRequestSigner( otsAction, config.getCredentials() ) );
        ec.setRetryStrategy( OTSRetryStrategyFactroy.getFactory().getStrategy( otsAction, parameters ) );

        if ( config.getServiceConfig().isEnableResponseContentMD5Checking() )
        {
            ec.getResponseHandlers().add( new OTSContentMD5ResponseHandler() );
        }
        // ec.getResponseHandlers().add( new OTSDeflateResponseHandler() );
        // ec.getResponseHandlers().add( new OTSErrorResponseHandler() );
        // ec.getResponseHandlers().add( new OTSValidationResponseHandler( config.getCredentials(), otsAction ) );

        return ec;
    }

    public OTSConfig getConfig()
    {
        return config;
    }

    public void setConfig( OTSConfig config )
    {
        this.config = config;
    }

}
